package cn.itcast.flink.streaming;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.flink.util.Collector;

/**
 * Author itcast
 * Date 2021/9/2 21:22
 * 将从socket读取的数据写入到 redis 中
 * 1. 接收数据并生成 wordcount
 * 2. 将结果写入到 redis 中
 * key : 单词
 * value : （单词,数量）
 */
public class RedisDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //读取socket数据流
        DataStreamSource<String> source = env.socketTextStream("192.168.88.", 9999);
        //做 wordcount
        source.flatMap(new RichFlatMapFunction<String, Tuple2<String,Long>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
                String[] words = value.split(" ");
                for (String word : words) {
                    out.collect(Tuple2.of(word,1L));
                }
            }
        }).keyBy(t->t.f0)
                .sum(1)
                .addSink(new RedisSink<Tuple2<String,Long>>(
                        new FlinkJedisPoolConfig.Builder()
                                .setHost("192.168.88.")
                                .setDatabase(0)
                                .build(),
                        new RedisMapper<Tuple2<String, Long>>() {
                            @Override
                            public RedisCommandDescription getCommandDescription() {
                                return new RedisCommandDescription(RedisCommand.HSET,"wordcount");
                            }

                            @Override
                            public String getKeyFromData(Tuple2<String, Long> data) {
                                return data.f0;
                            }

                            @Override
                            public String getValueFromData(Tuple2<String, Long> data) {
                                return data.f1.toString();
                            }
                        }
                ));
        //执行环境
        env.execute();
    }
}
